package net.wicp.tams.commons.zookeeper;

import java.util.List;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.Predicate;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import net.wicp.tams.commons.Conf;
import net.wicp.tams.commons.Result;
import net.wicp.tams.commons.callback.LockListener;
import net.wicp.tams.commons.zookeeper.opt.ZKLockOpt;
import net.wicp.tams.commons.zookeeper.opt.ZKUnLockOpt;

public class ZKLockManager {
	private static Logger logger = LoggerFactory.getLogger(Conf.class);
	private final String dir;
	private final LockListener lockListener;
	private final ZooKeeper zk = ZK.getInstance().getZk();

	public ZKLockManager(String dir) {
		this(dir, null);
	}

	public ZKLockManager(String dir, LockListener lockListener) {
		this.dir = dir;
		this.lockListener = lockListener;
		try {
			Stat stat = zk.exists(dir, false);
			if (stat != null) {
				return;
			}
			zk.create(dir, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
		} catch (Exception e) {
			logger.error("创建锁目录错误", e);
			throw new IllegalArgumentException("创建锁目录错误");
		}
	}

	/***
	 * 加分布式锁。
	 * 
	 * @param watcher
	 *            需要监听事件
	 * @return 数组，[0]锁的前一个路径 [1]锁的当前路径
	 * @throws KeeperException
	 * @throws InterruptedException
	 */
	public final synchronized String[] lock(Watcher watcher) throws KeeperException, InterruptedException {
		WatchProx watchertrue = new WatchProx(watcher, this);
		ZKLockOpt opt = new ZKLockOpt(dir);
		lockDo(opt, watchertrue);
		String[] retAry = new String[2];
		retAry[0] = tempRetObj[0] == null ? null : String.valueOf(tempRetObj[0]);
		retAry[1] = tempRetObj[1] == null ? null : String.valueOf(tempRetObj[1]);
		return retAry;
	}

	private Object[] tempRetObj;

	// 递归找前一个节点
	private void lockDo(ZKLockOpt opt, WatchProx watchertrue) throws KeeperException, InterruptedException {
		Result result = opt.execute();
		tempRetObj = result.getRetObjs();
		if (tempRetObj == null || tempRetObj[0] == null) {
			if (!watchertrue.watchIsNull()) {
				watchertrue.process(null);
			}
		} else {
			Stat stat = zk.exists(String.valueOf(tempRetObj[0]), watchertrue);
			if (stat == null) {
				lockDo(opt, watchertrue);
			}
		}
	}

	/***
	 * 得到当前锁的路径
	 * 
	 * @return
	 */
	public final String getCurPath() {
		try {
			List<String> names = zk.getChildren(dir, false);
			long sessionId = zk.getSessionId();
			final String prefix = String.format("x-%s-", sessionId);
			String curName = (String) CollectionUtils.find(names, new Predicate() {
				@Override
				public boolean evaluate(Object object) {
					String name = String.valueOf(object);
					return name.startsWith(prefix);
				}
			});
			return curName;
		} catch (Exception e) {
			logger.error("查看当前锁路径错误", e);
			return null;
		}

	}

	/***
	 * 解除指定的锁
	 * 
	 * @param path
	 * @return
	 */
	public final synchronized Result unlock(String path) {
		ZKUnLockOpt un = new ZKUnLockOpt(path);
		try {
			Result ret = un.execute();
			if (lockListener != null) {
				lockListener.lockReleased();
			}
			return ret;
		} catch (Exception e) {
			logger.error("解锁错误", e);
			return Result.getError("解锁错误:" + e.getMessage());
		}
	}

	/***
	 * 解除当前拥有的锁
	 * 
	 * @return
	 */
	public final synchronized Result unlock() {
		return unlock(getCurPath());
	}

	private final class WatchProx implements Watcher {
		private final Watcher watcher;
		private final ZKLockManager zKLockManager;

		private WatchProx(Watcher watcher, ZKLockManager zKLockManager) {
			this.watcher = watcher;
			this.zKLockManager = zKLockManager;
		}

		public boolean watchIsNull() {
			return watcher == null;
		}

		@Override
		public void process(WatchedEvent event) {
			// 在监听前一个节点事件时，如果前一个节点意外中止，但当前锁又没释放时要考虑
			if (event != null) {
				try {
					zKLockManager.lock(this);// this指WatchProx实例
				} catch (Exception e) {
					logger.error("再次锁错误", e);
				}
			}

			if (watcher != null) {
				watcher.process(event);
			}
			if (lockListener != null) {
				lockListener.lockAcquired();
			}
		}
	}

}
